package com.cjl.open.rxjava;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class RxJavaUsage {

    public static void main(String[] args) {
        final Disposable[] disposable = new Disposable[1];


        System.out.println("RxJavaUsage.main----------");
        Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                System.out.println("Observable RxJavaUsage.subscribe "+Thread.currentThread().getName());
                emitter.onNext("连载1");
                emitter.onNext("连载2");
                emitter.onNext("连载3");
                emitter.onComplete();
            }
        })

//        Observer<String> reader = ;


        .map(new Function<String, Object>() {

            @Override
            public Object apply(String s) throws Throwable {
                System.out.println("RxJavaUsage.apply "+Thread.currentThread().getName());
                return s;
            }
        })
//        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.newThread())
        .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                disposable[0] = d;
                System.out.println("RxJavaUsage.onSubscribe");
            }

            @Override
            public void onNext(@NonNull Object s) {
                System.out.println("Observer RxJavaUsage.onNext "+Thread.currentThread().getName() );
                if ("2".equals(s)){
                    disposable[0].dispose();
                    return;
                }
                System.out.println("RxJavaUsage.onNext "+s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("RxJavaUsage.onError "+e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("RxJavaUsage.onComplete");
            }
        });


//        ConnectableObservable<Long> connectableObservable = Observable.interval(200, TimeUnit.MILLISECONDS).publish();
//
//        connectableObservable.connect();
//
//        @NonNull Disposable disposable1 = connectableObservable.subscribe(new Consumer<Long>() {
//            @Override
//            public void accept(Long aLong) throws Throwable {
//                System.out.println("RxJavaUsage.accept  onNext first:"+aLong);
//            }
//        });
//
//        @NonNull Disposable disposable2 = connectableObservable.subscribe(new Consumer<Long>() {
//            @Override
//            public void accept(Long aLong) throws Throwable {
//                System.out.println("RxJavaUsage.accept  onNext second:"+aLong);
//            }
//        });
//
//        RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
//            @Override
//            public Observable apply(Observable observable) throws Throwable {
//                System.out.println("RxJavaUsage.apply ---");
//                return observable;
//            }
//        });
//        Observable.create(new ObservableOnSubscribe<Object>() {
//            @Override
//            public void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Throwable {
//
//            }
//        });


    }

}
